package com.atguigu.realtime.app.dwd;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd_OrderDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_OrderDetail().init(
                4001,
                2,
                "Dwd_OrderDetail"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //1.读取ods_db
        readOdsDb(tEnv, "Dwd_OrderDetail");

//        tEnv.sqlQuery("select * from ods_db").execute().print();

        Table orderDetail = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['course_id'] course_id, " +
                " `data`['course_name'] course_name, " +
                " `data`['order_id'] order_id, " +
                " `data`['user_id'] user_id, " +
                " `data`['origin_amount'] origin_amount," +
                " `data`['coupon_reduce'] coupon_reduce," +
                " `data`['final_amount'] final_amount," +
                " `data`['session_id'] session_id," +
                " `data`['create_time'] create_time," +
                " `data`['update_time'] update_time," +
                " ts," +
                " pt " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_detail' " +
                "and (" +
                " `type`='insert' " +
                ")");
        tEnv.createTemporaryView("order_detail", orderDetail);

        tEnv.executeSql("create table dwd_order_detail( " +
                "id string, " +
                "course_id string, " +
                "course_name string, " +
                "order_id string, " +
                "user_id string, " +
                "origin_amount string, " +
                "coupon_reduce string, " +
                "final_amount string, " +
                "session_id string, " +
                "create_time string, " +
                "update_time string, " +
                "ts bigint, " +
                "row_op_ts timestamp_ltz(3) " +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_ORDER_DETAIL));

        orderDetail.executeInsert("dwd_order_detail");







    }
}
